package com.shujia.advance;

import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

class MyBulkLoadMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, KeyValue>{
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, ImmutableBytesWritable, KeyValue>.Context context) throws IOException, InterruptedException, IOException {
        String line = value.toString();
        String[] strings = line.split("\t");
        //判断当不是脏数据的时候才开始处理
        if(strings.length>7 && !("\\N".equals(strings[1]))){
            String phoneNum = strings[0];
            String wg = strings[1];
            String city = strings[2];
            String qx = strings[3];
            String stayTime = strings[4];
            String startTime = strings[5];
            String endTime = strings[6];
            String date = strings[7];

            //手机号会重复 可以将手机号与开始进入时间一起作为rowkey
            String id = phoneNum+"_"+startTime;

            ImmutableBytesWritable new_id = new ImmutableBytesWritable(Bytes.toBytes(id));

            //将其他列封装成KeyValue对象
            //public KeyValue(final byte [] row, final byte [] family, final byte [] qualifier, final byte [] value)
            KeyValue value1 = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("info"), Bytes.toBytes("wg"), Bytes.toBytes(wg));
            KeyValue value2 = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("info"), Bytes.toBytes("city"), Bytes.toBytes(city));
            KeyValue value3 = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("info"), Bytes.toBytes("qx"), Bytes.toBytes(qx));
            KeyValue value4 = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("info"), Bytes.toBytes("stayTime"), Bytes.toBytes(stayTime));
            KeyValue value5 = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("info"), Bytes.toBytes("endTime"), Bytes.toBytes(endTime));
            KeyValue value6 = new KeyValue(Bytes.toBytes(id), Bytes.toBytes("info"), Bytes.toBytes("date"), Bytes.toBytes(date));

            context.write(new_id,value1);
            context.write(new_id,value2);
            context.write(new_id,value3);
            context.write(new_id,value4);
            context.write(new_id,value5);
            context.write(new_id,value6);
        }
    }
}
